package com.example.sse.service;

import com.alibaba.fastjson.JSON;
import com.example.sse.model.RealtimeData;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 * 集群通信服务
 * 基于Redis Pub/Sub实现集群间消息同步
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class ClusterCommunicationService {

    private final StringRedisTemplate redisTemplate;
    private final RedisMessageListenerContainer messageListenerContainer;
    private final SseConnectionManager sseConnectionManager;

    /**
     * Redis频道名称
     */
    private static final String SSE_CHANNEL = "sse:broadcast";

    /**
     * 初始化集群通信
     */
    @PostConstruct
    public void initClusterCommunication() {
        try {
            // 检查Redis连接是否可用
            if (messageListenerContainer.getConnectionFactory() == null) {
                log.info("Redis未配置或不可用，使用单机模式");
                return;
            }
            
            // 订阅Redis频道
            ChannelTopic topic = new ChannelTopic(SSE_CHANNEL);
            messageListenerContainer.addMessageListener((message, pattern) -> {
                try {
                    String jsonMessage = message.toString();
                    RealtimeData data = JSON.parseObject(jsonMessage, RealtimeData.class);
                    
                    // 只处理来自其他节点的消息（避免自己发送给自己）
                    if (!sseConnectionManager.getNodeId().equals(data.getNodeId())) {
                        log.debug("收到集群消息: {} from {}", data.getType(), data.getNodeId());
                        sseConnectionManager.broadcast(data);
                    }
                } catch (Exception e) {
                    log.error("处理集群消息失败: {}", e.getMessage());
                }
            }, topic);
            
            log.info("集群通信服务初始化完成，订阅频道: {}", SSE_CHANNEL);
        } catch (Exception e) {
            log.warn("集群通信初始化失败，将使用单机模式: {}", e.getMessage());
        }
    }

    /**
     * 向集群广播消息
     * @param data 要广播的数据
     */
    public void broadcastToCluster(RealtimeData data) {
        try {
            data.setNodeId(sseConnectionManager.getNodeId());
            
            // 检查Redis连接是否可用
            if (messageListenerContainer.getConnectionFactory() != null) {
                String jsonMessage = JSON.toJSONString(data);
                // 发送到Redis频道
                redisTemplate.convertAndSend(SSE_CHANNEL, jsonMessage);
                log.debug("消息已广播到集群: {}", data.getType());
            } else {
                log.debug("Redis不可用，仅在本地节点广播: {}", data.getType());
            }
            
            // 同时发送到本地连接
            sseConnectionManager.broadcast(data);
            
        } catch (Exception e) {
            log.error("广播消息到集群失败: {}", e.getMessage());
            // 如果Redis失败，至少保证本地连接能收到消息
            sseConnectionManager.broadcast(data);
        }
    }

    /**
     * 获取集群节点信息
     * @return 节点信息
     */
    public String getClusterNodeInfo() {
        return String.format("NodeId: %s, LocalConnections: %d", 
            sseConnectionManager.getNodeId(), 
            sseConnectionManager.getActiveConnectionCount());
    }
}